// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: taekwon

#pragma once

#include <atomic>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <memory>
#include <optional>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <boost/beast/websocket/stream_base.hpp>
#include <boost/beast/websocket/teardown.hpp>
#include <nlohmann/json.hpp>

#include "coin2/exchange/base/api_base/raw_subscriber.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/order/connection.pb.h"

namespace coin2::exchange::base::order {

class BaseOrderProcessor;

struct PrivateWsSubscriberSpec {
 public:
  PrivateWsSubscriberSpec(
      std::string_view host,
      std::string_view port,
      std::string_view path,
      std::string_view decompress,
      std::string_view topic,
      int ping_period_sec,
      int timeout_sec,
      bool handle_server_ping,
      bool use_ssl)
      : host(host),
        port(port),
        path(path),
        decompress(decompress),
        topic(topic),
        ping_period_sec(ping_period_sec),
        timeout_sec(timeout_sec),
        handle_server_ping(handle_server_ping),
        use_ssl(use_ssl) {}

  std::string host;
  std::string port;
  std::string path;
  std::string decompress;
  std::string topic;
  int ping_period_sec;
  int timeout_sec;
  bool handle_server_ping;
  bool use_ssl;
};

class BasePrivateWsSubscriberPolicy {
 public:
  using MarketExchangeApi = coin2::exchange::base::market::MarketExchangeApi;
  using OrderConnectionConfig = coin2::exchange::order::OrderConnectionConfig;

  BasePrivateWsSubscriberPolicy(
      const MarketExchangeApi& mea,
      const OrderConnectionConfig& conn_config,
      const std::set<std::string>& native_symbols);
  std::vector<std::string> GetSubscriptionMsgs() const;
  std::string GetAuthMsg() const;
  std::string GetPingMsg() const;
  std::optional<std::string> MakePingMsgFromServer(const std::string& server_ping) const;
  bool IsSuccessAuthResp(const std::string& resp) const;
  bool IsConnectionErrorMsg(const std::string&) const;
  const OrderConnectionConfig& GetOrderConnectionConfig() const { return conn_config_; }
  auto GetMockPingMsg() const -> std::optional<std::string>;

  nlohmann::json ws_config_json_;
  MarketExchangeApi mea_;
  OrderConnectionConfig conn_config_;
  const std::set<std::string> native_symbols_;
};

class BasePrivateWsConnection : public std::enable_shared_from_this<BasePrivateWsConnection> {
 public:
  using OrderConnectionConfig = coin2::exchange::order::OrderConnectionConfig;

  BasePrivateWsConnection(
      const PrivateWsSubscriberSpec& spec,
      std::unique_ptr<BasePrivateWsSubscriberPolicy> policy,
      std::function<void()> fail_callback,
      boost::asio::io_context* ioc,
      const MarketExchangeApi& mea);

  virtual ~BasePrivateWsConnection();

  void RunAsync();

  std::string Decompress(const std::string& s) const;

  void
  Fail(boost::asio::yield_context* yield, boost::system::error_code ec, const std::string& what);

  void Subscribe(boost::asio::yield_context yield);
  void Run(boost::asio::yield_context yield);

  void SetProcessor(BaseOrderProcessor* processor) { processor_ = processor; }
  auto processor() { return processor_; }

  void SetListenKey(const std::string& listen_key) { listen_key_ = listen_key; }
  bool HasListenKey() const { return (!listen_key_.empty()); }

  void SetUserId(const std::string& user_id) { user_id_ = user_id; }
  bool HasUserId() const { return (!user_id_.empty()); }

  void SetBullet(const std::vector<std::string>& bullet_info) { bullet_info_ = bullet_info; }
  bool HasBullet() const { return (bullet_info_.size() == 5); }

  bool IsValidSpec() const { return !spec_.host.empty() && !spec_.port.empty(); }

  bool CheckIfNeedReconnect(std::string_view msg_view);
  void ForceResetConnection() { reset_connection_ = true; }
  void SetLastRecvMsgTs(int64_t ts) { last_recv_msg_ts_ = ts; }
  auto MakeMockPingMsg() -> std::optional<std::string>;
  void Monitor(boost::asio::yield_context yield);

 protected:
  std::function<void()> fail_callback_;
  std::atomic<bool> running_;
  std::atomic<bool> reset_connection_;
  std::atomic<bool> subscribe_done_;

  PrivateWsSubscriberSpec spec_;
  std::unique_ptr<BasePrivateWsSubscriberPolicy> policy_;

  typedef typename boost::beast::websocket::stream<
      boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>
      WebsocketSsl;
  typedef typename boost::beast::websocket::stream<boost::asio::ip::tcp::socket> WebsocketNoSsl;

  boost::asio::io_context* ioc_;
  boost::asio::io_context::strand strand_;
  boost::asio::ssl::context ctx_;
  boost::asio::ip::tcp::resolver resolver_;
  bool use_ssl_;

  std::unique_ptr<WebsocketSsl> ws_ssl_;
  std::unique_ptr<WebsocketNoSsl> ws_nossl_;

  boost::beast::multi_buffer buffer_;
  const MarketExchangeApi mea_;
  BaseOrderProcessor* processor_;

  std::string listen_key_;
  std::string user_id_;
  std::vector<std::string> bullet_info_;

  int64_t last_recv_msg_ts_ = 0;
};

}  // namespace coin2::exchange::base::order
